package com.edu.learning.concurrent.sample.mode1;

import java.time.Duration;
import java.time.LocalTime;
import java.util.concurrent.*;

/**
 * @author ll
 * @Date 2017/10/24 10:29
 */
public class ConcurrentLinkedQueueTest {
    private static ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<Integer>();
    private final static int TOTAL_COUNT = 100000;
    private final static int THREAD_COUNT = 2;
    private final static int THREAD_POOL_COUNT = 32;
    /**
     * CountDownLatch同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待
     */
    private static CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT);

    public static void Produce() {
        for (int i = 0; i < TOTAL_COUNT; i ++) {
            queue.offer(i);
        }
    }

    public static void main(String[] args) {
        LocalTime start = LocalTime.now();
        ExecutorService executorService = new ThreadPoolExecutor(
                THREAD_POOL_COUNT,
                THREAD_POOL_COUNT,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(1024),
                new ThreadPoolExecutor.CallerRunsPolicy());
        ConcurrentLinkedQueueTest.Produce();
        for (int i = 0; i < THREAD_COUNT; i ++) {
            executorService.submit(new Consume());
        }
        try {
            //使得主线程(main)阻塞直到latch.countDown()为零才继续执行
            countDownLatch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("cost time "
                + (Duration.between(LocalTime.now(), start)) + "ms");
        countDownLatch.countDown();
    }

    static class Consume implements Runnable {
        @Override
        public void run() {
            //while (queue.size() > 0) { /// cost time PT-10.008Sms
            while (!queue.isEmpty()) { /// cost time PT-0.576Sms
                System.out.println(queue.poll());
            }
            countDownLatch.countDown();
        }
    }
}
